package com.project.RocketRoute.executor;

import com.project.RocketRoute.model.RocketRouteBean;
import io.netty.channel.DefaultChannelId;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.RPCHook;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.context.annotation.Bean;

import java.util.List;
import java.util.Map;
import java.util.TreeMap;

@SpringBootConfiguration
public class RocketConsumerExecutor {

    @Value("${rocketmq.namesrv}")
    private String namesrvAddr;
    @Value("${rocketmq.group}")
    private String groupName;
    @Value("${rocketmq.threadMin}")
    private int threadMin;
    @Value("${rocketmq.threadMax}")
    private int threadMax;
    @Value("${rocketmq.batchMaxSize}")
    private int batchMaxSize;

    @Value("${rocketmq.accessKey:}")
    private String accessKey;

    @Value("${rocketmq.secretKey:}")
    private String secretKey;

    @Value("${rocketmq.groups:}")
    private String groups;

    private Map<String, DefaultMQPushConsumer> consumers = new TreeMap<>();


    @Autowired
    private RocketMqListenerConsumer rocketMqListenerConsumer;

//    private DefaultMQPushConsumer getProducer() {
//        if (StringUtils.isNotBlank(accessKey)) {
//            AclClientRPCHook acl = new AclClientRPCHook(new SessionCredentials(accessKey, secretKey));
//            return new DefaultMQPushConsumer(groupName, acl, new AllocateMessageQueueAveragely(), true, null);
//        }
//
//        return new DefaultMQPushConsumer(groupName);
//    }

    private DefaultMQPushConsumer getProducer(String name) {
        if (StringUtils.isNotBlank(accessKey)) {
            AclClientRPCHook acl = new AclClientRPCHook(new SessionCredentials(accessKey, secretKey));
            return new DefaultMQPushConsumer(name, acl, new AllocateMessageQueueAveragely(), true, null);
        }

        return new DefaultMQPushConsumer(name);
    }

    public DefaultMQPushConsumer bind(String name) {
        if (StringUtils.isBlank(name)) {
            name = groupName;
        }

        if (consumers.containsKey(name)) {
            return consumers.get(name);
        }

        DefaultMQPushConsumer consumer = getProducer(name);
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumeThreadMin(threadMin);
        consumer.setConsumeThreadMax(threadMax);
        consumer.registerMessageListener(rocketMqListenerConsumer);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.setMessageModel(MessageModel.CLUSTERING); //集群模式
        consumer.setConsumeMessageBatchMaxSize(batchMaxSize);
        System.out.println("===============>create rocket: " + rocketMqListenerConsumer);
        Map<String, List<RocketRouteBean>> beans = RocketRouteSetting.getBeans();
        try
        {
            System.out.println("===============>rocket items: " + beans.entrySet().size());
            for (Map.Entry<String, List<RocketRouteBean>> map : beans.entrySet()) {
                for (RocketRouteBean bean : map.getValue()){
                    String grp_id = StringUtils.isBlank(bean.getRouteMethod().groupId()) ? groupName : bean.getRouteMethod().groupId();
                    if (name.equals(grp_id)) {
                        System.out.println(String.format("======================>bind rocket topic: %s; tags: %s",
                                bean.getRouteMethod().topic(), bean.getRouteMethod().tags()));
                        consumer.subscribe(bean.getRouteMethod().topic(), bean.getRouteMethod().tags());
                    }
                }
            }

            DefaultChannelId.newInstance();
            consumer.start();
            consumers.put(name, consumer);
            System.out.println("===============>success rocket: " + name);
        } catch (Exception e){
            System.out.println("===============>error rocket: " + name);
            e.printStackTrace();
        }
        return consumer;
    }

    @Bean
    public DefaultMQPushConsumer rocketConsumer(){
        bind(groupName);
        if (StringUtils.isNotBlank(groups)) {
            String[] group_list = groups.split(",");
            for (String s : group_list) {
                bind(s);
            }
        }

        return consumers.get(groupName);
    }
}
